package com.zhu.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhu.bean.TableProcess;
import com.zhu.config.JDBCConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;


import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

import java.util.Map;
import java.util.Set;

/**
 * 主流连接广播配置流处理函数
 */
public class DIMTableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {


    //jdbc
    private Connection connection;

    MapStateDescriptor<String,TableProcess> mapStateDescriptor;

    public DIMTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    /**
     * 数据主流处理
     * 读取状态中维度表信息
     * 根据sinkColumn 过滤主流数据 filterColumn
     * 补充SinkTable字段
     * 写入到Phoenix
     * @param jsonObject
     * @param readOnlyContext
     * @param collector
     * @throws Exception
     */

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext readOnlyContext,
                               Collector<JSONObject> collector) throws Exception{
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        System.out.println(jsonObject.toJSONString());
        TableProcess tableProcess = broadcastState.get(jsonObject.getString("table")); //null ?
        //过滤字段
        if(tableProcess != null) {
            filterColumn(jsonObject.getJSONObject("data"), tableProcess.getSinkColumns());
            jsonObject.put("sinkTable",tableProcess.getSinkTable());
            collector.collect(jsonObject);
            //
        }else {
            System.out.println("未查询到该维表");
        }
    }

    /**
     * 广播处理
     * 获取并解析数据 方便主流操作
     * 校验表是否存在 不存在在Phoenix if not exist
     * 构建状态 维度表信息 写入状态
     * @param value
     * @param context
     * @param collector
     * @throws Exception
     */

    //{"op":"READ","before":{},"after":{"sink_table":"bb","source_table":"aa","sink_extend":"xxx",
    // "sink_pk":"dd","sink_columns":"cc"},"db":"gmall-flink-config","tableName":"table_process"}

    /*
    {"op":"READ","before":{},"after":{"sink_table":"dim_base_tradmark","source_table":"base_trademark","sink_extend":"",
    "sink_pk":"id","sink_columns":"id,tm_name"},
    "db":"gmall-flink-config","ts_ms":"1679661776716","tableName":"table_process"}
     */
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context context,
                                        Collector<JSONObject> collector) throws Exception {
        //提取after 封装成process
        JSONObject jsonObject = JSON.parseObject(value);
        String after = jsonObject.getString("after");
        TableProcess tableProcess = JSON.parseObject(after,TableProcess.class);

        //校验建表
        BroadcastState<String, TableProcess> dimTableBroadcastState = context.getBroadcastState(mapStateDescriptor);

        dimTableBroadcastState.put(tableProcess.getSourceTable(),tableProcess);  //封装好状态数据
        checkTable(tableProcess.getSinkTable(),tableProcess.getSinkColumns(),tableProcess.getSinkPK(),tableProcess.getSinkExtend());

    }

    /**
     * 校验并建表 create table if not exists db.tn(aa varchar primary key,bb varchar ,cc varchar) xxx
     * @param sinkTable   Phoenix表名
     * @param sinkColumns   Phoenix表字段
     * @param sinkPK        Phoenix表主键
     * @param sinkExtend    Phoenix扩展字段
     *                     sql 拼接
     */
    private void checkTable(String sinkTable, String sinkColumns, String sinkPK, String sinkExtend) {
        //处理特殊字段  没有扩展字段

        PreparedStatement preparedStatement = null;
        try {
            if ("".equals(sinkPK) || sinkPK == null) {
                sinkPK = "id";
            }
            if (sinkExtend == null) {
                sinkExtend = "";
            }

            StringBuilder createTableSql = new StringBuilder("CREATE TABLE IF NOT EXISTS ")
                    .append(JDBCConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");
            String[] columns = sinkColumns.split(",");
            for (int i = 0; i < columns.length; i++) {
                String column = columns[i];
                if (sinkPK.equals(column)) {
                    createTableSql.append(column).append(" VARCHAR PRIMARY KEY");
                } else {
                    createTableSql.append(column).append(" VARCHAR");
                }
                //判断是否是最后一个字段
                if (i < columns.length - 1) {
                    createTableSql.append(",");
                }
            }
            createTableSql.append(")").append(sinkExtend);   //扩展字段

            //编译sql
            System.out.println("create table >>>" + createTableSql);
            preparedStatement = connection.prepareStatement(createTableSql.toString());
            preparedStatement.execute();
        }catch (SQLException sqlException){
            sqlException.printStackTrace();
            throw new RuntimeException("create table false" +sinkTable);  //程序会停掉
        }finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();  //编译资源关闭
                }catch (SQLException sqlException){
                    sqlException.printStackTrace();
                }
            }
        }
    }

    /**
     * 过滤数据
     * @param data
     * @param sinkColumns
     */
    private void filterColumn(JSONObject data, String sinkColumns) {
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !sinkColumns.contains(next.getKey()));

    }

    /**
     * @param parameters 配置
     * @throws Exception
     * 创建JDBC 连接
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(JDBCConfig.PHOENIX_SERVER);
    }

    @Override
    public void close() throws Exception {
        //close connection
    }
}
